#pragma once
#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
#include <openssl/ssl.h>
#include <openssl/opensslv.h>

#include "logger.hpp"

namespace im{

    class MQClient{
    public:
        using MessageCallBack = std::function<void(const char*, size_t)>;
        using ptr = std::shared_ptr<MQClient>;
        MQClient(const std::string& user, const std::string& pwd, const std::string& host)
            :_loop(EV_DEFAULT), _handler(AMQP::LibEvHandler(_loop)){
            
            std::string url = "amqp://" + user + ":" + pwd + "@" + host + "/";
            AMQP::Address address(url);

            _conn = std::make_unique<AMQP::TcpConnection>(&_handler, address);
            _channel = std::make_unique<AMQP::TcpChannel>(_conn.get());
            
            // 新启一个线程跑这个功能，因为是它是阻塞的
            _loop_thread = std::thread([&]{
                ev_run(_loop, 0);
            });

            // 初始化异步事件监听器
            ev_async_init(&_async_watcher, watcher_callback);
            ev_async_start(_loop, &_async_watcher);
        }
        ~MQClient(){
            LOG_INFO("MQClient destructor called...");
            // 确保线程安全
            if (_conn) {
                _conn->close();
            }

            if (_channel) {
                _channel->close();
            }

            // 发送异步事件通知事件循环退出
            ev_async_send(_loop, &_async_watcher);

            // 等待事件循环线程退出
            if (_loop_thread.joinable()) {
                _loop_thread.join();
                LOG_INFO("Event loop thread joined.");
            } else {
                LOG_CRITICAL("Event loop thread not joinable.");
            }

            // 清理异步事件监听器
            ev_async_stop(_loop, &_async_watcher);

            // 置空指针
            _loop = nullptr;
        }

        void declare_component(const std::string& exchange,
                                const std::string& queue, 
                                const std::string& routing_key = "routing_key",
                                AMQP::ExchangeType exchange_type = AMQP::ExchangeType::direct){
            _channel->declareExchange(exchange, exchange_type)
                .onSuccess([&]{
                    LOG_INFO("{} 交换机创建成功！", exchange);
                })
                .onError([](const char* msg){
                    LOG_CRITICAL("声明交换机失败：{}", msg);
                    exit(0);
                });

            _channel->declareQueue(queue)
                .onSuccess([&](){
                    LOG_INFO("{} 队列创建成功！", queue);
                })
                .onError([](const char *msg) {
                    LOG_CRITICAL("声明队列失败：{}", msg);
                    exit(0);
                });

            _channel->bindQueue(exchange, queue, routing_key)
                .onSuccess([&](){
                    LOG_INFO("{} - {} - {} 绑定成功！", exchange, queue, routing_key);
                })
                .onError([&](const char *msg) {
                    LOG_CRITICAL("{} - {} 绑定失败：{}", exchange, queue, msg);
                    exit(0);
                });
        }

        bool publish(const std::string& exchange, 
                    const std::string& msg, 
                    const std::string& routing_key = "routing_key"){
            bool ret = _channel->publish(exchange, routing_key, msg);
            LOG_DEBUG("向交换机 {}-{} 发布消息！", exchange, routing_key);
            if (ret == false) {
                    LOG_CRITICAL("{} 发布消息失败：", exchange);
                    return false;
            }
            return true;
        }

        void consume(const std::string& queue, const MessageCallBack &cb){
            LOG_DEBUG("开始订阅 {} 队列消息！", queue);
            _channel->consume(queue)
                .onReceived([this, cb](const AMQP::Message &message, uint64_t deliveryTag, bool redelivered){
                    cb(message.body(), message.bodySize());
                    _channel->ack(deliveryTag);
                })
                .onError([&](const char* msg){
                    LOG_CRITICAL("订阅 {} 队列消息失败: {}", queue, msg);
                    exit(0);
                });
        }
    private:
        static void watcher_callback(struct ev_loop *loop, ev_async *watcher, int32_t revents) {
            ev_break(loop, EVBREAK_ALL);
        }
        
    private:
        struct ev_loop* _loop;
        AMQP::LibEvHandler _handler;
        std::unique_ptr<AMQP::TcpConnection> _conn;
        std::unique_ptr<AMQP::TcpChannel> _channel;
        std::thread _loop_thread;
        struct ev_async _async_watcher; // 异步事件监听器
    };
}